package com.alibaba.citrus.cr.mn.order.center.facade.service.impl;

import com.alibaba.citrus.cr.common.enums.ordercenter.AutoFulfillmentOrderStatusCirculationEnum;
import com.alibaba.citrus.cr.common.enums.ordercenter.AutoFulfillmentOrderStatusEnum;
import com.alibaba.citrus.cr.common.exception.FunctionException;
import com.alibaba.citrus.cr.common.lock.SaleRedisLock;
import com.alibaba.citrus.cr.mn.order.center.autofulfillmentorder.service.AutoFulfillmentOrderBizQueryService;
import com.alibaba.citrus.cr.mn.order.center.autofulfillmentorder.service.AutoFulfillmentOrderBizWriteService;
import com.alibaba.citrus.cr.mn.order.center.autofulfillmentorder.service.AutoFulfillmentOrderService;
import com.alibaba.citrus.cr.mn.order.center.common.constants.RedisLockConstants;
import com.alibaba.citrus.cr.mn.order.center.facade.function.common.OrderAtomQueryService;
import com.alibaba.citrus.cr.mn.order.center.facade.function.convert.AnalyzeReasonToCirculationStatusConvert;
import com.alibaba.citrus.cr.mn.order.center.forecastorder.ForecastOrderBizDomainService;
import com.alibaba.cz.base.tool.log.Log;
import com.alibaba.cz.base.tool.page.PageResult;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.lattice2.epoch.db.service.CustomSequenceAccessor;
import com.alibaba.lattice2.epoch.sdk.model.Result;
import com.epoch.app.crb2btradecenter.enums.OrderStatus;
import com.epoch.app.crforecastsales.domian.cfsbcautofulfillmentorderrecords.dto.ConsumerAnalyzeMessageRequest;
import com.epoch.app.crforecastsales.domian.cfsbcautofulfillmentorderrecords.dto.ConsumerPushMessageRequest;
import com.epoch.app.crforecastsales.domian.cfsbcautofulfillmentorderrecords.dto.QueryForListRequest;
import com.epoch.app.crforecastsales.domian.cfsbcautofulfillmentorderrecords.model.CfsBcAutoFulfillmentOrderRecords;
import com.epoch.app.crforecastsales.domian.cfsbcautofulfillmentorderrecords.service.CfsBcAutoFulfillmentOrderRecordsService;
import com.epoch.app.crforecastsales.model.enums.ForecastDealStatusSEO;
import com.epoch.app.mnordercenter.model.dto.*;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author baizhuo
 * @version 1.0
 * @date 2023/8/22 17:47
 */
@Service
@Primary
public class CfsBcAutoFulfillmentOrderRecordsBizServiceImpl implements CfsBcAutoFulfillmentOrderRecordsService {

    private static final Log log = Log.getLogger(CfsBcAutoFulfillmentOrderRecordsBizServiceImpl.class);

    @Resource
    private AutoFulfillmentOrderBizQueryService autoFulfillmentOrderBizQueryService;

    @Resource
    private AutoFulfillmentOrderBizWriteService autoFulfillmentOrderBizWriteService;

    @Resource
    private AutoFulfillmentOrderService autoFulfillmentOrderService;

    @Resource
    private ForecastOrderBizDomainService forecastOrderBizDomainService;

    @Resource(name="salesOrderAtomQueryService")
    private OrderAtomQueryService salesOrderAtomQueryService;

    @Resource
    private CustomSequenceAccessor customSequenceAccessor;

    @Override
    public Result<Void> consumerAnalyzeMessage(ConsumerAnalyzeMessageRequest consumerAnalyzeMessageRequest) {
        log.info("消费批量分析消息开始, param:{}", JSON.toJSONString(consumerAnalyzeMessageRequest));
        try {
            String message = consumerAnalyzeMessageRequest.getMessage();
            CfsBcAutoFulfillmentOrderRecords cfsBcAutoFulfillmentOrderRecords = JSONObject.parseObject(message, CfsBcAutoFulfillmentOrderRecords.class);
            //请求幂等
            String lockKey = RedisLockConstants.FORECAST_ORDER_BIZ_PREFIX + cfsBcAutoFulfillmentOrderRecords.getFromSys() + "#" + cfsBcAutoFulfillmentOrderRecords.getAnalyzeBatchNo();
            try (SaleRedisLock reqLock = new SaleRedisLock(RedisLockConstants.CACHE_CODE, lockKey, RedisLockConstants.TIMEOUT_MSECS, RedisLockConstants.EXPIRE_MSECS, RedisLockConstants.SLEEP_MILLIS)) {
                if (!reqLock.lock()) {
                    log.error("CfsBcAutoFulfillmentOrderRecordsBizServiceImpl#consumerAnalyzeMessage 分布式锁获取失败 lockKey={}", lockKey);
                    throw new FunctionException("OC-00-000-01-99-002");
                }
                QueryForListRequest queryForListRequest = QueryForListRequest.builder().analyzeBatchNo(cfsBcAutoFulfillmentOrderRecords.getAnalyzeBatchNo()).build();
                List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordsListBeforeFilter = autoFulfillmentOrderBizQueryService.queryForList(queryForListRequest);
                //过滤符合条件的记录
                List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordsList = cfsBcAutoFulfillmentOrderRecordsListBeforeFilter.stream().filter(cfs -> AutoFulfillmentOrderStatusEnum.PROCESSING.getCode() == cfs.getStatus() && AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_FAILED.getCode() < cfs.getCirculationStatus() && AutoFulfillmentOrderStatusCirculationEnum.ALL_PASS.getCode() >= cfs.getCirculationStatus()).collect(Collectors.toList());
                if (CollectionUtils.isEmpty(cfsBcAutoFulfillmentOrderRecordsList)) {
                    log.info("消费批量分析消息#查询自动履约订单为空 result is null, param:{}, result:{}", JSON.toJSONString(queryForListRequest), JSON.toJSONString(cfsBcAutoFulfillmentOrderRecordsListBeforeFilter));
                    return Result.success(null);
                }
                Map<String, CfsBcAutoFulfillmentOrderRecords> salesOrderNoLineNoMap = cfsBcAutoFulfillmentOrderRecordsList.stream().collect(Collectors.toMap(autoFulfillmentOrderRecords -> autoFulfillmentOrderRecords.getOrderBizNo()+ "_" + autoFulfillmentOrderRecords.getOrderLineBizNo(), Function.identity(), (k1, k2) -> k2));
                ForecastOrderBizQueryAnalysisRequest forecastOrderBizQueryAnalysisRequest = new ForecastOrderBizQueryAnalysisRequest();
                forecastOrderBizQueryAnalysisRequest.setFromSys(cfsBcAutoFulfillmentOrderRecords.getFromSys());
                forecastOrderBizQueryAnalysisRequest.setBusinessType(cfsBcAutoFulfillmentOrderRecords.getBusinessType());
                forecastOrderBizQueryAnalysisRequest.setBusinessUnitCode(cfsBcAutoFulfillmentOrderRecords.getBusinessUnitCode());
                forecastOrderBizQueryAnalysisRequest.setBatchNo(cfsBcAutoFulfillmentOrderRecords.getAnalyzeBatchNo());
                String sequenceNo = customSequenceAccessor.getNextStringSequence("cr_forecast_sales", "sequenceNo");
                forecastOrderBizQueryAnalysisRequest.setReqNo(sequenceNo);
                log.info("消费批量分析消息#查询预报子单开始 param:{}", JSON.toJSONString(forecastOrderBizQueryAnalysisRequest));
                Result<List<ForecastOrderBizQueryAnalysisResponse>> queryAnalyzeResult = forecastOrderBizDomainService.queryAnalysis(forecastOrderBizQueryAnalysisRequest);
                log.info("消费批量分析消息#查询预报子单结束 result:{}", JSON.toJSONString(queryAnalyzeResult));
                boolean needSendBack = !queryAnalyzeResult.isSuccess() ? true : queryAnalyzeResult.getResult().stream().anyMatch(forecastOrderBizQueryAnalysisResponse -> forecastOrderBizQueryAnalysisResponse.getDealStatus().equals(ForecastDealStatusSEO.ANALYSIS_ING.getStatus()));
                if (needSendBack) {
                    //接口返回失败或返回正在分析中，抛异常，利用mq自动重试，重试到一定次数进入死信队列，防止无限重试
                    log.info("消费批量分析消息#查询预报子单返回失败或分析未完成, success={}", queryAnalyzeResult.isSuccess());
                    throw new FunctionException("OC-04-001-01-16-031");
                }
                //查询分析结果,根据返回的分析结果更新到自动履约订单表中；
                List<String> orderBizNoList = new ArrayList<>(cfsBcAutoFulfillmentOrderRecordsList.stream().map(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo).collect(Collectors.toSet()));
                List<PageQueryForecastDetailResponse> forecastDetailResponseList = this.pageQueryLine(orderBizNoList, cfsBcAutoFulfillmentOrderRecords);
                List<CfsBcAutoFulfillmentOrderRecords> analyzeAllPass = new ArrayList<>();
                //更新circulationStatus状态
                if (CollectionUtils.isNotEmpty(forecastDetailResponseList)) {
                    Set<String> salesOrderNoLineNoSet = forecastDetailResponseList.stream().map(detailResponse -> detailResponse.getSalesOrderNo() + "_" + detailResponse.getLineNum()).collect(Collectors.toSet());
                    List<String> salesOrderNoLineNoList = new ArrayList<>(salesOrderNoLineNoMap.keySet());
                    boolean returnLessThanQuery = salesOrderNoLineNoList.stream().anyMatch(key -> !salesOrderNoLineNoSet.contains(key));
                    if (returnLessThanQuery) {
                        log.info("结果比预期少, query size:{}, return size:{}", salesOrderNoLineNoMap.size(), forecastDetailResponseList.size());
                        throw new FunctionException("OC-04-001-01-16-032");
                    }
                    List<CfsBcAutoFulfillmentOrderRecords> needUpdateList = new ArrayList<>();
                    for (PageQueryForecastDetailResponse response : forecastDetailResponseList) {
                        log.info("转换状态开始 CirculationStatus param:" + JSON.toJSONString(response));
                        Integer circulationStatus = AnalyzeReasonToCirculationStatusConvert.convert(response.getAnalysisReason(), response.getActivityConfilctFlag());
                        log.info("转换状态结束 CirculationStatus result:" + circulationStatus);
                        if (null == circulationStatus) {
                            log.info("转换状态结果为空, param:{}", JSON.toJSONString(response));
                            //测试环境出现过forecastOrderBizDomainService.queryAnalysis返回分析完成，但forecastOrderBizDomainService.pageQueryLine返回analysisReason为空的情况，需要重试等待analysisReason被赋值
                            throw new FunctionException("OC-04-001-01-16-032");
                        }
                        String salesOrderNoLineNo = response.getSalesOrderNo() + "_" + response.getLineNum();
                        if (salesOrderNoLineNoMap.containsKey(salesOrderNoLineNo)) {
                            CfsBcAutoFulfillmentOrderRecords cfsBcAutoFulfillmentOrderRecordsDB = salesOrderNoLineNoMap.get(salesOrderNoLineNo);
                            cfsBcAutoFulfillmentOrderRecordsDB.setCirculationStatus(circulationStatus);
                            cfsBcAutoFulfillmentOrderRecordsDB.setGmt_modified(new Date());
                            if (!circulationStatus.equals(AutoFulfillmentOrderStatusCirculationEnum.ALL_PASS.getCode())) {
                                cfsBcAutoFulfillmentOrderRecordsDB.setStatus(AutoFulfillmentOrderStatusEnum.FAIL.getCode());
                                cfsBcAutoFulfillmentOrderRecordsDB.setFailReason(AutoFulfillmentOrderStatusCirculationEnum.of(circulationStatus).getDesc());
                            } else {
                                //分析结果是"完全通过"，重置failReason
                                cfsBcAutoFulfillmentOrderRecordsDB.setStatus(AutoFulfillmentOrderStatusEnum.PROCESSING.getCode());
                                cfsBcAutoFulfillmentOrderRecordsDB.setFailReason("");
                                needUpdateList.add(cfsBcAutoFulfillmentOrderRecordsDB);
                            }
                            autoFulfillmentOrderBizWriteService.update(cfsBcAutoFulfillmentOrderRecordsDB);
                        }
                    }
                    if (CollectionUtils.isNotEmpty(needUpdateList)) {
                        //重新查一下，把status=0、circulation_status=28的重新捞出来一起确认
                        List<String> needQueryAgainOrderBizNoList = needUpdateList.stream().map(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo).distinct().collect(Collectors.toList());
                        QueryForListRequest queryAgainRequest = QueryForListRequest.builder().orderBizNos(needQueryAgainOrderBizNoList).build();
                        List<CfsBcAutoFulfillmentOrderRecords> queryAgainResult = autoFulfillmentOrderBizQueryService.queryForList(queryAgainRequest);
                        Map<String, List<CfsBcAutoFulfillmentOrderRecords>> queryAgainOrderBizNoMap = queryAgainResult.stream().collect(Collectors.groupingBy(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo));
                        for (Map.Entry<String, List<CfsBcAutoFulfillmentOrderRecords>> mapEntry : queryAgainOrderBizNoMap.entrySet()) {
                            //如果是一主多子，分析结果可能出现某子单完全满足、另外某子单不是完全满足，此时认为主单维度整体失败
                            Optional<CfsBcAutoFulfillmentOrderRecords> any = mapEntry.getValue().stream().filter(cfsOrder -> AutoFulfillmentOrderStatusEnum.FAIL.getCode() == cfsOrder.getStatus()
                                    || (AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_SUCCEED.getCode() < cfsOrder.getCirculationStatus() && AutoFulfillmentOrderStatusCirculationEnum.ALL_PASS.getCode() > cfsOrder.getCirculationStatus()
                                    && AutoFulfillmentOrderStatusCirculationEnum.CREDIT_NOT_ENOUGH.getCode() != cfsOrder.getCirculationStatus())).findAny();
                            if (any.isPresent()) {
                                for (CfsBcAutoFulfillmentOrderRecords fail : mapEntry.getValue()) {
                                    fail.setStatus(AutoFulfillmentOrderStatusEnum.FAIL.getCode());
                                    fail.setGmt_modified(new Date());
                                    autoFulfillmentOrderBizWriteService.update(fail);
                                }
                            }  else {
                                boolean allAllowConfirm = mapEntry.getValue().stream().allMatch(cbRecord -> AutoFulfillmentOrderStatusEnum.PROCESSING.getCode() == cbRecord.getStatus()
                                        && (AutoFulfillmentOrderStatusCirculationEnum.ALL_PASS.getCode() == cbRecord.getCirculationStatus() || AutoFulfillmentOrderStatusCirculationEnum.CREATE_SALE_ORDER_FAILED.getCode() == cbRecord.getCirculationStatus()));
                                if (allAllowConfirm) {
                                    analyzeAllPass.addAll(mapEntry.getValue());
                                }
                            }
                        }
                    }
                    if (CollectionUtils.isNotEmpty(analyzeAllPass)) {
                        log.info("分析完成，即将开始确认.");
                        autoFulfillmentOrderService.filterOrderAndBatchConfirm(analyzeAllPass);
                    } else {
                        //没有一个分析通过，统一发送Kafka消息
                        log.info("consumerAnalyzeMessage no order analyze pass, send kafka message.");
                        autoFulfillmentOrderService.sendMessage(cfsBcAutoFulfillmentOrderRecords.getBatchNo(), cfsBcAutoFulfillmentOrderRecords.getFromSys(), cfsBcAutoFulfillmentOrderRecords.getBusinessType(), cfsBcAutoFulfillmentOrderRecords.getBusinessUnitCode());
                    }
                } else {
                    //接口返回失败，抛异常重试
                    log.info("查询预报子单接口执行失败，无返回值, result:{}", JSON.toJSONString(forecastDetailResponseList));
                    throw new FunctionException("OC-04-001-01-16-032");
                }
            } catch (Exception e) {
                log.error("消费批量分析消息异常", e);
                throw e;
            }
        } catch (Exception e1) {
            log.error("消费批量分析消息失败", e1);
            throw e1;
        }
        log.info("消费批量分析消息结束");
        return Result.success(null);
    }

    private List<PageQueryForecastDetailResponse> pageQueryLine(List<String> orderBizNoList, CfsBcAutoFulfillmentOrderRecords cfs) {
        List<PageQueryForecastDetailResponse> result = new ArrayList<>();
        boolean hasNext = true;
        int pageNo = 1;
        PageQueryForecastLineRequest pageQueryForecastLineRequest = new PageQueryForecastLineRequest();
        pageQueryForecastLineRequest.setSalesOrderNoList(orderBizNoList);
        pageQueryForecastLineRequest.setSize(100);
        pageQueryForecastLineRequest.setFromSys(cfs.getFromSys());
        pageQueryForecastLineRequest.setBusinessType(cfs.getBusinessType());
        pageQueryForecastLineRequest.setBusinessUnitCode(cfs.getBusinessUnitCode());
        while (hasNext) {
            pageQueryForecastLineRequest.setPage(pageNo);
            log.info("消费批量分析消息#查询预报子单开始 param:{}", JSON.toJSONString(pageQueryForecastLineRequest));
            Result<List<PageQueryForecastDetailResponse>> forecastOrderDetailQueryResult = forecastOrderBizDomainService.pageQueryLine(pageQueryForecastLineRequest);
            log.info("消费批量分析消息#查询预报子单结束 success:{}", forecastOrderDetailQueryResult.isSuccess());
            if (forecastOrderDetailQueryResult.isSuccess() && CollectionUtils.isNotEmpty(forecastOrderDetailQueryResult.getResult())) {
                result.addAll(forecastOrderDetailQueryResult.getResult());
            }
            if (forecastOrderDetailQueryResult.getTotal() <= result.size()) {
                hasNext = false;
            }
            pageNo++;
            if (pageNo >= 20) {
                break;
            }
        }
        return result;
    }


    @Override
    public Result<Void> consumerPushMessage(ConsumerPushMessageRequest consumerPushMessageRequest) {
        log.info("消费push消息开始, param:{}", JSON.toJSONString(consumerPushMessageRequest));
        try {
            String message = consumerPushMessageRequest.getMessage();
            CfsBcAutoFulfillmentOrderRecords cfsBcAutoFulfillmentOrderRecords = JSONObject.parseObject(message, CfsBcAutoFulfillmentOrderRecords.class);
            //请求幂等
            String lockKey = RedisLockConstants.FORECAST_ORDER_BIZ_PREFIX + cfsBcAutoFulfillmentOrderRecords.getFromSys() + "#" + cfsBcAutoFulfillmentOrderRecords.getConfirmBatchNo();
            try (SaleRedisLock reqLock = new SaleRedisLock(RedisLockConstants.CACHE_CODE, lockKey, RedisLockConstants.TIMEOUT_MSECS, RedisLockConstants.EXPIRE_MSECS, RedisLockConstants.SLEEP_MILLIS)) {
                if (!reqLock.lock()) {
                    log.error("AutoFulfillmentOrderServiceImpl#consumerPushMessage 分布式锁获取失败 lockKey={}", lockKey);
                    throw new FunctionException("OC-00-000-01-99-002");
                }
                //根据预报订单确认批次号查询
                QueryForListRequest queryForListRequest = QueryForListRequest.builder().confirmBatchNo(cfsBcAutoFulfillmentOrderRecords.getConfirmBatchNo()).build();
                queryForListRequest.setFromSys(cfsBcAutoFulfillmentOrderRecords.getFromSys());
                queryForListRequest.setBusinessType(cfsBcAutoFulfillmentOrderRecords.getBusinessType());
                queryForListRequest.setBusinessUnitCode(cfsBcAutoFulfillmentOrderRecords.getBusinessUnitCode());
                List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordsListBeforeFilter = autoFulfillmentOrderBizQueryService.queryForList(queryForListRequest).stream().filter(record -> record.getCirculationStatus() > AutoFulfillmentOrderStatusCirculationEnum.CREATE_FORECAST_ORDER_FAILED.getCode()).collect(Collectors.toList());
                List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordsList = cfsBcAutoFulfillmentOrderRecordsListBeforeFilter.stream().filter(cfsRecord -> AutoFulfillmentOrderStatusEnum.PROCESSING.getCode() == cfsRecord.getStatus() && AutoFulfillmentOrderStatusCirculationEnum.ALL_PASS.getCode() < cfsRecord.getCirculationStatus() && AutoFulfillmentOrderStatusCirculationEnum.PUSH_TO_SAP_SUCCEED.getCode() != cfsRecord.getCirculationStatus()).collect(Collectors.toList());
                if (CollectionUtils.isEmpty(cfsBcAutoFulfillmentOrderRecordsList)) {
                    log.info("没有满足push条件的数据 by param:{}", JSON.toJSONString(queryForListRequest));
                    return Result.success(null);
                }
                Map<String, List<CfsBcAutoFulfillmentOrderRecords>> orderBizNoMap = cfsBcAutoFulfillmentOrderRecordsList.stream().collect(Collectors.groupingBy(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo));
                List<String> orderBizNoList = new ArrayList<>(cfsBcAutoFulfillmentOrderRecordsList.stream().map(CfsBcAutoFulfillmentOrderRecords::getOrderBizNo).collect(Collectors.toSet()));
                List<SalesOrderAtomQueryResponse> saleOrderList = this.listPageQuerySaleOrder(orderBizNoList, cfsBcAutoFulfillmentOrderRecordsList.get(0));
                boolean needSendBack = CollectionUtils.isEmpty(saleOrderList) || saleOrderList.stream().anyMatch(salesOrderAtomQueryResponse -> StringUtils.isEmpty(salesOrderAtomQueryResponse.getOrderStatus()) || (StringUtils.isNotEmpty(salesOrderAtomQueryResponse.getOrderStatus()) && ((Integer.valueOf(salesOrderAtomQueryResponse.getOrderStatus()).equals(OrderStatus.BEFORE_SUBMIT.getCode()) || Integer.valueOf(salesOrderAtomQueryResponse.getOrderStatus()).equals(OrderStatus.PUSHING.getCode())))));
                if (needSendBack) {
                    //接口返回失败或有销售订单状态是待提交（待提交实际表示未推送）或已推送（已推送表示推送了但是sap未返回结果）,抛异常，消息重试
                    log.info("listPageForInner内部查询失败或push失败");
                    throw new FunctionException("OC-04-001-01-16-032");
                }
                if (!CollectionUtils.isEmpty(saleOrderList)) {
                    for (SalesOrderAtomQueryResponse salesOrderAtomQueryResponse : saleOrderList) {
                        if (orderBizNoMap.containsKey(salesOrderAtomQueryResponse.getPreOrderNo())) {
                            List<CfsBcAutoFulfillmentOrderRecords> cfsBcAutoFulfillmentOrderRecordsListDB = orderBizNoMap.get(salesOrderAtomQueryResponse.getPreOrderNo());
                            for (CfsBcAutoFulfillmentOrderRecords autoFulfillmentOrderRecords : cfsBcAutoFulfillmentOrderRecordsListDB) {
                                log.info("销售订单返回值 orderBizNo:{}, OrderStatus:{}, salesOrderNo:{}, outOrderNo:{}", salesOrderAtomQueryResponse.getPreOrderNo(), salesOrderAtomQueryResponse.getOrderStatus(), salesOrderAtomQueryResponse.getSalesOrderNo(), salesOrderAtomQueryResponse.getOutOrderNo());
                                Map<String, SalesOrderAtomQueryDetailResponse> sourceLineNoMap = salesOrderAtomQueryResponse.getOrderDetails().stream().collect(Collectors.toMap(SalesOrderAtomQueryDetailResponse::getSourceLineNum, Function.identity(), (k1, k2) -> k2));
                                if (StringUtils.isNotEmpty(salesOrderAtomQueryResponse.getSalesOrderNo()) || StringUtils.isNotEmpty(salesOrderAtomQueryResponse.getOutOrderNo())) {
                                    JSONObject jsonObject = StringUtils.isEmpty(autoFulfillmentOrderRecords.getFeatures()) ? new JSONObject() : JSONObject.parseObject(autoFulfillmentOrderRecords.getFeatures());
                                    if (!jsonObject.containsKey("salesOrderNo") || Objects.isNull(jsonObject.get("salesOrderNo")) || StringUtils.isEmpty(jsonObject.get("salesOrderNo").toString())) {
                                        jsonObject.put("salesOrderNo", salesOrderAtomQueryResponse.getSalesOrderNo());
                                    }
                                    if (!jsonObject.containsKey("outOrderNo") || Objects.isNull(jsonObject.get("outOrderNo")) || StringUtils.isEmpty(jsonObject.get("outOrderNo").toString())) {
                                        jsonObject.put("outOrderNo", salesOrderAtomQueryResponse.getOutOrderNo());
                                    }
                                    autoFulfillmentOrderRecords.setFeatures(jsonObject.toJSONString());
                                }
                                if (Integer.valueOf(salesOrderAtomQueryResponse.getOrderStatus()).equals(OrderStatus.STAY_PUSH.getCode())) {
                                    //待推送实际指的是推送失败
                                    if (sourceLineNoMap.containsKey(autoFulfillmentOrderRecords.getSourceOrderLineNo())) {
                                        autoFulfillmentOrderRecords.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.PUSH_TO_SAP_FAILED.getCode());
                                        autoFulfillmentOrderRecords.setStatus(AutoFulfillmentOrderStatusEnum.FAIL.getCode());
                                        JSONObject responseFeature = StringUtils.isEmpty(salesOrderAtomQueryResponse.getBizFeatures()) ? new JSONObject() : JSONObject.parseObject(salesOrderAtomQueryResponse.getBizFeatures());
                                        autoFulfillmentOrderRecords.setFailReason(responseFeature.containsKey("sapReturnMessage") ? responseFeature.get("sapReturnMessage").toString() : "推送失败");
                                    }
                                } else if (Integer.parseInt(salesOrderAtomQueryResponse.getOrderStatus()) > OrderStatus.BEFORE_SUBMIT.getCode()) {
                                    if (sourceLineNoMap.containsKey(autoFulfillmentOrderRecords.getSourceOrderLineNo())) {
                                        //sap订单号及sap订单行号是通过两个接口更新，可能存在sap订单号有值，但是sap订单行号为空的情况，此时需再次重试
                                        if (StringUtils.isEmpty(sourceLineNoMap.get(autoFulfillmentOrderRecords.getSourceOrderLineNo()).getOutLineNum())) {
                                            log.info("sap子单行号为空.");
                                            throw new FunctionException("OC-04-001-01-16-032");
                                        }
                                        autoFulfillmentOrderRecords.setCirculationStatus(AutoFulfillmentOrderStatusCirculationEnum.PUSH_TO_SAP_SUCCEED.getCode());
                                        autoFulfillmentOrderRecords.setStatus(AutoFulfillmentOrderStatusEnum.SUCCESS.getCode());
                                        autoFulfillmentOrderRecords.setFailReason("");
                                        //外部单号当前就是指sap订单号，外部订单行号当前就是指sap订单行号
                                        autoFulfillmentOrderRecords.setSapOrderNo(salesOrderAtomQueryResponse.getOutOrderNo());
                                        autoFulfillmentOrderRecords.setSapOrderLineNo(sourceLineNoMap.get(autoFulfillmentOrderRecords.getSourceOrderLineNo()).getOutLineNum());
                                    }
                                }
                                autoFulfillmentOrderRecords.setGmt_modified(new Date());
                                autoFulfillmentOrderBizWriteService.update(autoFulfillmentOrderRecords);
                            }
                        }
                    }
                    //统一发送Kafka消息
                    log.info("AutoFulfillmentOrderServiceImpl#consumerPushMessage send kafka message");
                    autoFulfillmentOrderService.sendMessage(cfsBcAutoFulfillmentOrderRecords.getBatchNo(), cfsBcAutoFulfillmentOrderRecords.getFromSys(), cfsBcAutoFulfillmentOrderRecords.getBusinessType(), cfsBcAutoFulfillmentOrderRecords.getBusinessUnitCode());
                }
            } catch (Exception e) {
                log.error("消费push消息失败 ", e);
                throw e;
            }
        } catch (Exception e1) {
            log.info("消费push消息异常", e1);
            throw e1;
        }
        log.info("消费push消息结束");
        return Result.success(null);
    }

    private List<SalesOrderAtomQueryResponse> listPageQuerySaleOrder(List<String> orderBizNoList, CfsBcAutoFulfillmentOrderRecords cfs) {
        List<SalesOrderAtomQueryResponse> result = new ArrayList<>();
        boolean hasNext = true;
        int pageNo = 1;
        SalesOrderAtomQueryListRequest salesOrderAtomQueryListRequest = new SalesOrderAtomQueryListRequest();
        salesOrderAtomQueryListRequest.setSize(100);
        salesOrderAtomQueryListRequest.setFromSys(cfs.getFromSys());
        salesOrderAtomQueryListRequest.setBusinessType(cfs.getBusinessType());
        salesOrderAtomQueryListRequest.setBusinessUnitCode(cfs.getBusinessUnitCode());
        salesOrderAtomQueryListRequest.setPreOrderNo(orderBizNoList);
        while (hasNext) {
            salesOrderAtomQueryListRequest.setPage(pageNo);
            log.info("批量查询销售订单开始 param:{}", JSON.toJSONString(salesOrderAtomQueryListRequest));
            PageResult<List<SalesOrderAtomQueryResponse>> saleOrderQueryResult = salesOrderAtomQueryService.listPageForInner(salesOrderAtomQueryListRequest);
            log.info("批量查询销售订单结束 result:{}", JSON.toJSONString(saleOrderQueryResult));
            if (CollectionUtils.isNotEmpty(saleOrderQueryResult.getContent())) {
                result.addAll(saleOrderQueryResult.getContent());
            }
            if (saleOrderQueryResult.getTotal() <= result.size()) {
                hasNext = false;
            }
            pageNo++;
            if (pageNo >= 20) {
                break;
            }
        }
        return result;
    }
}
